package com.study.bigdata.spark.streaming

import com.study.bigdata.spark.streaming.SparkStreaming11_Req1_BlackList.AdClickData
import com.study.bigdata.spark.util.JDBCUtil
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

object SparkStreaming12_Req2 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc =new StreamingContext(sparkConf,Seconds(3))
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",//消费者组ID
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //位置策略，采集结点和计算结点如何做匹配？自己匹配：框架匹配
      ConsumerStrategies.Subscribe[String, String](Set("atguiguNew"), kafkaPara) //消费者策略,主题：atguiguNew
    )
    val adClickData = kafkaDataDS.map {
      kafkaData => {
        val data = kafkaData.value()
        val datas = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    }
    val reducDS = adClickData.map(
      data => {
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        val day = sdf.format(new Date((data.ts.toLong)))
        val area = data.area
        val city = data.city
        val ad = data.ad
        ((day, area, city, ad), 1)
      }
    ).reduceByKey(_ + _)

    reducDS.foreachRDD(
      rdd =>{
        rdd.foreachPartition(
          iter =>{
            val coon =JDBCUtil.getConnection
            val pstat =coon.prepareStatement(
              """
                |insert into area_city_ad_count (dt,area,city,adid,count)
                |values (?,?,?,?,?)
                |on DUPLICATE KEY
                |UPDATE count=count + ?
                |""".stripMargin)
            iter.foreach{
              case ((day, area, city, ad), sum) =>{
                pstat.setString(1,day)
                pstat.setString(2,area)
                pstat.setString(3,city)
                pstat.setString(4,ad)
                pstat.setInt(5,sum)
                pstat.setInt(6,sum)
                pstat.executeUpdate()
              }
            }
            pstat.close()
            coon.close()
          }
        )
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }
  //广告点击数据样例类
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
